package bigdata.scala.spark.streaming.ycfx.tgmodule

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.util

import bigdata.java.framework.spark.kafka.{SKProcessor, SimpleSparkKafka}
import bigdata.java.framework.spark.util.client.HBaseClient
import bigdata.scala.spark.streaming.ycfx.model._
import bigdata.scala.spark.streaming.ycfx.util.GetCurrentTimeUtil
import com.alibaba.fastjson.JSON
import org.apache.hadoop.hbase.client.Result
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.api.java.JavaInputDStream


object TgModule {
  def getLineLossStatus(args: Array[String]): Unit = {
    // val Array(appname,topics,duration,maxnum) = args

    val simpleSparkKafka = new SimpleSparkKafka(args)

    simpleSparkKafka.execute(new SKProcessor {
      override def process(directStream: JavaInputDStream[ConsumerRecord[String, String]]): Unit = {
        val stream = directStream.inputDStream
        var lossRate: String  = ""

        stream.foreachRDD{rdd=>
          if(!rdd.isEmpty()){
            val baseData = rdd.map(kafka => JSON.parseObject(kafka.value()))
              .map((json => {
                val cityOrgNo = json.getString("CITY_ORG_NO")
                val cityOrgName = json.getString("CITY_ORG_NAME")
                val countyOrgNo = json.getString("COUNTY_ORG_NO")
                val countyOrgName = json.getString("COUNTY_ORG_NAME")
                val orgNo = json.getString("ORG_NO")
                val orgName = json.getString("ORG_NAME")
                val tgId = json.getString("TG_ID")
                val tgNo = json.getString("TG_NO")
                val tgName = json.getString("TG_NAME")
                val transformerType = json.getString("TRANSFORMER_TYPE")
                val lineId = json.getString("LINE_ID")
                var lineName = json.getString("LINE_NAME")
                val runDate = json.getString("RUN_DATE")
                val runStatusCode = json.getString("RUN_STATUS_CODE")
                val pubPrivFlag = json.getString("PUB_PRIV_FLAG")
                val tgCap = json.getString("TG_CAP")
                val collectTime = json.getString("COLLECT_TIME")
                val supplyQuantity = json.getString("SUPPLY_QUANTITY")
                val sellQuantity = json.getString("SELL_QUANTITY")
                //损失电量=供电量-售电量
                val quantityLoss =(supplyQuantity.toDouble - sellQuantity.toDouble).toString
                //获得更新时间
                val updateTime = GetCurrentTimeUtil.getNowDate()
                val createTime = GetCurrentTimeUtil.getNowDate()
                //线损计算
                if(supplyQuantity==0.00){
                  lossRate = "0.00"
                }else {
                  //线损=损失电量 / 供电量 *100  精度保留两位
                  lossRate = (quantityLoss.toDouble/supplyQuantity.toDouble *100).formatted("%.2f")


                }
                //实体类set值
                TgModel(cityOrgNo,cityOrgName,countyOrgNo,countyOrgName,orgNo,orgName,tgId,tgNo,tgName,transformerType,lineId,lineName,runDate,runStatusCode,pubPrivFlag,tgCap,collectTime,supplyQuantity,sellQuantity,quantityLoss,lossRate,createTime,updateTime)
              }))
            //落地HBase
            baseData.foreachPartition { partitionOfRecods =>
              partitionOfRecods.foreach(t => {
                val list = new util.ArrayList[util.Map[String,String]]()
                val map = new util.HashMap[String,String]
                map.put(" ",t.cityOrgNo)
                map.put("cityOrgName",t.cityOrgName)
                map.put("countyOrgNo",t.countyOrgNo)
                map.put("countyOrgName",t.countyOrgName)
                map.put("orgNo",t.orgNo)
                map.put("orgName",t.orgName)
                map.put("tgId",t.tgId)
                map.put("tgNo",t.tgNo)
                map.put("tgName",t.tgName)
                map.put("transformerType",t.transformerType)
                map.put("lineId",t.lineId)
                map.put("lineName",t.lineName)
                map.put("runDate",t.runDate)
                map.put("runStatusCode",t.runStatusCode)
                map.put("pubPrivFlag",t.pubPrivFlag)
                map.put("tgCap",t.tgCap)
                map.put("collectTime",t.collectTime)
                map.put("supplyQuantity",t.supplyQuantity)
                map.put("sellQuantity",t.sellQuantity)
                map.put("quantityLoss",t.quantityLoss)
                map.put("updateTime",t.updateTime)
                map.put("createTime",t.createTime)
                map.put("lossRate",t.lossRate)
                list.add(map)
                HBaseClient.getInstance().put("","","",list)

              })
            }

          }
        }
      }
    }).start()


  }



  //根据台区id 查台区下用户
  def findTgUser(baseData:RDD[UserStatus]): Map[String,String] ={
      var tgName:String = null
      var userName:String = null
      baseData.foreachPartition(data=>{
      data.foreach(t=>{
        val result: Result = HBaseClient.getInstance().get("",s"${t.tgNo}")
        tgName = ""
        userName = ""
      })
    })
    Map[String,String](tgName,userName)
  }

}
